package spring.module.disruptor;

import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;

import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;

import spring.module.disruptor.log.LogEvent;
import spring.module.disruptor.log.LogEventFactory;
import spring.module.disruptor.log.LogEventHandler;
import spring.module.disruptor.message.MessageConsumer;
import spring.module.disruptor.message.MessageEventFactory;
import spring.module.disruptor.message.MessageModel;

/**
 * 配置
 * 
 * @author Lucky Yang
 * 
 */
@Configuration
public class Config {

    @Bean
    public RingBuffer<MessageModel> messageModelRingBuffer() {
        // 定义用于事件处理的线程池，
        ThreadFactory threadFactory = new CustomizableThreadFactory("disruptor-pool-");

        // 指定事件工厂
        MessageEventFactory factory = new MessageEventFactory();

        // 指定ringbuffer字节大小，必须为2的N次方（能将求模运算转为位运算提高效率），否则将影响效率
        int bufferSize = 1024 * 256;

        Disruptor<MessageModel> disruptor = new Disruptor<>(factory, bufferSize, threadFactory);

        // 设置事件业务处理器---消费者
        MessageConsumer[] consumers = new MessageConsumer[] { new MessageConsumer(), new MessageConsumer() };
        disruptor.handleEventsWith(consumers);

        // 启动disruptor线程
        disruptor.start();

        // 获取ringbuffer环，用于接取生产者生产的事件
        return disruptor.getRingBuffer();
    }

    @Bean
    public RingBuffer<LogEvent> logModelRingBuffer() {
        // 创建 LogEvent 的工厂
        LogEventFactory logEventFactory = new LogEventFactory();
        // Disruptor 的 RingBuffer 缓存大小
        int bufferSize = 1024 * 1024;
        // 生产者的线程工厂
        ThreadFactory threadFactory = new ThreadFactory() {
            final AtomicInteger threadNum = new AtomicInteger(0);

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "LogEventThread" + " [#" + threadNum.incrementAndGet() + "]");
            }
        };
        // 实例化 Disruptor
        Disruptor<LogEvent> disruptor = new Disruptor<>(logEventFactory, bufferSize, threadFactory);

        // 设置事件业务处理器---消费者
        LogEventHandler[] handlers = new LogEventHandler[] { new LogEventHandler()};
        disruptor.handleEventsWith(handlers);

        // 启动disruptor线程
        disruptor.start();

        // 获取ringbuffer环，用于接取生产者生产的事件
        return disruptor.getRingBuffer();
    }
}
